package com.example.cloud.config.redis;

import java.util.Collection;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.lang3.builder.HashCodeBuilder;
import org.apache.ibatis.cache.Cache;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;

import com.example.cloud.util.spring.SpringContextUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * MyBatis基于Redis的缓存实现
 *
 * @author tongyufu
 * @since 2018年11月16日 下午5:11:52
 */
@Slf4j
public class MyBatisRedisCache implements Cache {

    /**
     * 默认缓存过期时间：30分钟
     */
    public static final Long    DEFAULT_EXPIRE_MINUTES = 30L;
    private final ReadWriteLock readWriteLock          = new ReentrantReadWriteLock();
    private String              id;
    private int                 idHashCode;
    private RedisTemplate       redisTemplate;

    /**
     * 构造函数
     *
     * @param id ID
     */
    public MyBatisRedisCache(String id) {
        this.id = id;
        idHashCode = generateHashCode(id);
    }

    @Override
    public String getId() {
        return id;
    }

    @Override
    public void putObject(Object key, Object value) {
        String wrapKey = warpKey(key);
        ValueOperations valueOperations = getCache().opsForValue();
        HashOperations hashOperations = getCache().opsForHash();
        Boolean absent = valueOperations.setIfAbsent(wrapKey, value);
        getCache().expire(wrapKey, DEFAULT_EXPIRE_MINUTES, TimeUnit.MINUTES);

        //MyBatis默认的Key太长，占用redis空间太多，对Key做了HashCode后，当发生hash碰撞时，将值转成Hash类型
        if (!absent) {
            //获取值的方法的类型不一致时会报错，所以把Hash的Key加个后缀
            String hashKey = wrapKey + ":h";
            //第一个Key已经转换成HashCode，无法得知原始值是什么，所以用转换后的Key作为第一个值的Key
            hashOperations.put(hashKey, wrapKey, valueOperations.get(wrapKey));
            hashOperations.put(hashKey, key.toString(), value);
            valueOperations.set(wrapKey, "_not_absent_", DEFAULT_EXPIRE_MINUTES, TimeUnit.MINUTES);
            getCache().expire(hashKey, DEFAULT_EXPIRE_MINUTES, TimeUnit.MINUTES);
        }
    }

    @Override
    public Object getObject(Object key) {
        String wrapKey = warpKey(key);
        Object value = getCache().opsForValue().get(wrapKey);
        if (value == null) {
            return null;
        }
        if ("_not_absent_".equals(value.toString())) {
            String hashKey = wrapKey + ":h";
            value = getCache().opsForHash().get(hashKey, key.toString());
            if (value == null) {
                value = getCache().opsForHash().get(hashKey, wrapKey);
            }
        }
        if (value instanceof Collection) {
            return new CopyOnWriteArrayList((Collection) value);
        }
        return value;
    }

    @Override
    public Object removeObject(Object key) {
        return getCache().delete(warpKey(key));
    }

    @Override
    public void clear() {
        Set<String> keys = getCache().keys(idHashCode + ":*");
        getCache().delete(keys);
        getCache().execute((RedisCallback) redisCallback -> {
            redisCallback.flushDb();
            return null;
        });
    }

    @Override
    public int getSize() {
        return (int) getCache().execute((RedisCallback) redisCallback -> redisCallback.dbSize().intValue());
    }

    @Override
    public ReadWriteLock getReadWriteLock() {
        return readWriteLock;
    }

    private RedisTemplate<String, Object> getCache() {
        if (redisTemplate == null) {
            redisTemplate = SpringContextUtils.getBean("redisTemplate", RedisTemplate.class);
        }
        return redisTemplate;
    }

    private String warpKey(Object key) {
        return idHashCode + ":" + generateHashCode(key);
    }

    private int generateHashCode(Object key) {
        return new HashCodeBuilder(17, 37).append(key).toHashCode();
    }

}
